spb/autoscholar

1import { listEvents } from "@autoscholar/core";2import { redact } from "@autoscholar/observability";3import { z } from "zod";4import { handler, json, parseQuery } from "@/lib/server/api";5import { db } from "@/lib/server/db";6import { requireAdmin } from "@/lib/server/session";78export const runtime = "nodejs";9export const dynamic = "force-dynamic";1011/**12 * Reconnectable event stream. `?after=<seq>` (or `Last-Event-ID`) resumes; `?stream=1` returns Server-Sent Events,13 * otherwise a JSON page. Events are redacted before leaving the server.14 */15export const GET = handler(async (req, { params }) => {16  await requireAdmin(req);17  const { id } = (await params) as { id: string };18  const q = parseQuery(19    req.url,20    z.object({21      after: z.coerce.number().int().min(0).default(0),22      stream: z.string().optional(),23      limit: z.coerce.number().int().min(1).max(1000).default(500),24    }),25  );26  const lastId = Number(req.headers.get("last-event-id") ?? 0);27  let after = Math.max(q.after, Number.isFinite(lastId) ? lastId : 0);28  if (!q.stream) {29    const events = await listEvents(db(), id, after, q.limit);30    return json({ events: redact(events), lastSeq: events.at(-1)?.seq ?? after });31  }32  const encoder = new TextEncoder();33  let closed = false;34  const stream = new ReadableStream<Uint8Array>({35    async start(controller) {36      const send = (data: unknown, seq?: number, event = "run.event") =>37        controller.enqueue(38          encoder.encode(39            `${seq !== undefined ? `id: ${seq}\n` : ""}event: ${event}\ndata: ${JSON.stringify(data)}\n\n`,40          ),41        );42      send({ connected: true, after }, undefined, "hello");43      const started = Date.now();44      while (!closed && Date.now() - started < 10 * 60_000) {45        try {46          const events = await listEvents(db(), id, after, 200);47          for (const ev of events) {48            send(redact(ev), ev.seq);49            after = ev.seq;50          }51          if (events.length === 0) controller.enqueue(encoder.encode(": keepalive\n\n"));52        } catch (e) {53          send({ error: (e as Error).message }, undefined, "error");54        }55        await new Promise((r) => setTimeout(r, 1500));56      }57      controller.close();58    },59    cancel() {60      closed = true;61    },62  });63  return new Response(stream, {64    headers: {65      "content-type": "text/event-stream; charset=utf-8",66      "cache-control": "no-store, no-transform",67      connection: "keep-alive",68      "x-accel-buffering": "no",69    },70  });71});72